#ifndef BLOCK_STATUS_H
#define BLOCK_STATUS_H

#include <list>
#include <map>
#include <algorithm>

#include <sdfs/storage/Block.h>
#include <sdfs/config/Json.h>

#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>

#include <muduo/base/Mutex.h>
#include <muduo/base/Atomic.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/TcpConnection.h>

// #include <sdfs/config/BlockStatusConfig.h>

using namespace sdfs::common;
using namespace boost;

using namespace muduo;
using namespace muduo::net;

const int BUFFER_SIZE = 102400;

namespace sdfs
{
namespace storage
{

typedef std::vector<PiecePtr> PieceTable;

class BlockStatus : public noncopyable
{
public:
	BlockStatus(int index, BlockPtr block, long blocklimit):
		index_(index), block_(block),
		blocklimit_(blocklimit), freesize_(0)
	{

	}
	~BlockStatus()
	{

	}

	PieceTable* getPieces()
	{
		return &freePieces_;
	}

	void setIndex(int idx)
	{
		index_ = idx;
	}

	int getIndex()
	{
		return index_;
	}

	void setBlock(BlockPtr block)
	{
		block_ = block;
	}

	BlockPtr getBlock()
	{
		return block_;
	}

	void setUsed(long capacity)
	{
		used_ = capacity;
	}

	long getUsed()
	{
		return used_;
	}

	long getFreeSize()
	{
		return freesize_;
	}

	void setFreeSize(long freesize)
	{
		freesize_ = freesize;
	}

	long getCapacity()
	{
		return blocklimit_;
	}

	void addFreePiece(PiecePtr piece)
	{
		//TODO
		// PiecePtr p = std::find(
		// 	freePieces_.begin(), freePieces_.end(), *piece);
		// if(p != NULL)
		// 	return;
		PieceTable::iterator it;
		freesize_ += piece->getLength();
		for (it = freePieces_.begin(); it!= freePieces_.end() && *it < piece; it++)
		{
			if((*it)->connected(piece))
			{
				(*it)->enlargeLength(piece->getLength());
				PieceTable::iterator it2 = it++;
				PiecePtr tmp = *it;
				if((*it2)->connected(tmp))
				{
					(*it2)->enlargeLength(tmp->getLength());
					freePieces_.erase(it2);
				}
				return;
			}else if( piece->connected(*it))
			{
				piece->enlargeLength( (*it)->getLength() );
				*it = piece;
			}
		}
		if(it == freePieces_.end())
		{
			freePieces_.push_back(piece);
			std::sort(freePieces_.begin(), freePieces_.end(), SortFunctor<Piece>());
		}
		//TODO: sync to disk
	}
	
	/*
	* 
	*/
	PiecePtr getPiece(long length)
	{
		// assert(status_ == WRITING);
		PiecePtr p;
		LOG_DEBUG << "getPiece: " << length <<" from: "
			<< block_->getIndex()<<"-"<<block_->getLastPos();
		// if(lockToWrite())
		// {
		//1. seek from deleted piece
		{
			MutexLockGuard guard(freeMutex_);
			PieceTable::iterator it;
			for (it = freePieces_.begin(); it != freePieces_.end(); it++)
			{
				if((*it)->isFeet(length))
				{
					PiecePtr p = (*it)->split(length);
					if((*it) == p)
						freePieces_.erase(it);
					break;
				}
			}
		}
		//2. if no empty piece exist, check writing piece, and
		//   lock a unused piece
		if(p == NULL)
		{
			MutexLockGuard guard(appendMutex_);
			long lastpos = block_->getLastPos();
			if( blocklimit_ > lastpos+length)
			{
				p.reset(new Piece(index_, lastpos, length));
				long offset = lastpos+length;
				block_->updateLastPos(offset);
			}
		}
			
		//TODO: defragmentation 
		return p;
	}

	/**
	* end of write
	**/
	// void unlockToRead()
	// {
	// 	MutexLockGuard guard(mutex_);
	// 	assert(status_ == WRITING);
	// 	status_ = READING;
	// }

	// void assertWriting()
	// {
	// 	assert(status_ == WRITING);
	// }

	// void assertReading()
	// {
	// 	assert(status_ == READING);
	// }
private:
	/**
	* try to switch status to write, success returns true
	**/
	// bool lockToWrite()
	// {
	// 	MutexLockGuard guard(mutex_);
	// 	if(status_ == WRITING)
	// 		return false;
	// 	status_ = WRITING;
	// 	return true;
	// }
	
private:

	int index_;
	BlockPtr block_;
	PieceTable freePieces_;
	muduo::MutexLock freeMutex_;
	// std::list<PiecePtr> usedPieces_;
	long used_;
	long blocklimit_;
	muduo::MutexLock appendMutex_;
	
	//muduo::MutexLock writemutex_;

	long freesize_;
	
};

typedef shared_ptr<BlockStatus> BlockStatusPtr;

class BlockStatusTable
{
public:
	BlockStatusTable(string& blockpath, int blocknum, long maxblocksize, int readersize)
	{
		init(blockpath, blocknum, maxblocksize, readersize);
	}
	~BlockStatusTable()
	{

	}

	void delData(const PiecePtr piece)
	{
		BlockStatusPtr status = table_[piece->getIndex()];
		status->addFreePiece(piece);
	}
	
	PiecePtr lockPiece(long length)
	{
		PiecePtr piece;
		BlockStatusPtr status;
		do
		{
			status = table_[index_.addAndGet(1) % size_.get()+1];
			piece = status->getPiece(length);
			piece->rewind();
		}while(piece == NULL);
		return piece;
	}

	bool SaveData(PiecePtr& piece, muduo::net::Buffer* buffer)
	{
		assert(piece != NULL);
		BlockStatusPtr status = table_[piece->getIndex()];
		// status->assertWriting();
		BlockPtr block = status->getBlock();
		WriteOnlyFilePtr writer = block->getWriter();
		writer->seekFromBeginning(piece->getPointer());
		ulong remain = piece->getLength() - (piece->getPointer()-piece->getOffset());
		ulong nread = 0;
		ulong n = 0;
		
		while(buffer->readableBytes() > 0 && nread < buffer->readableBytes())
		{
			n = writer->write(buffer->peek()+nread, buffer->readableBytes());
			buffer->retrieve(n);
			nread += n;
		}
		if(nread == remain)
		{
			// status->unlockToRead();
			return true;
		}
		piece->slipPointer(nread);
		return false;
	}

	ReadOnlyFilePtr getReader(PiecePtr& piece)
	{
		BlockTable::iterator it = table_.find(piece->getIndex());
		assert(it != table_.end());
		BlockPtr block = it->second->getBlock();
		return block->getReader();
	}


	void sendData(PiecePtr piece, TcpConnectionPtr& conn)
	{
		BlockTable::iterator it = table_.find(piece->getIndex());
		assert(it != table_.end());
		BlockPtr block = it->second->getBlock();
		ReadOnlyFilePtr reader = block->getReader();
		reader->seekFromBeginning(piece->getOffset());
		long n = 0;
		long rest = piece->getLength();
		char buf[BUFFER_SIZE];
		while(rest > BUFFER_SIZE)
		{
			n += reader->read(buf, BUFFER_SIZE);
			conn->send(buf, n);
			rest -= n;
		}
		reader->read(buf, rest);
		conn->send(buf, rest);
		conn->shutdown();
	}
private:

	void init(string& blockpath, int blocknum, long maxblocksize, int readersize)
	{
		char pathbuf[256];
		size_.getAndSet(0);
		for (int i = 1; i <= blocknum; ++i)
		{
			snprintf(pathbuf, 256, "%s/%d", blockpath.c_str(), i);
			muduo::string p(pathbuf);
			BlockPtr block(new Block(i, p, readersize, maxblocksize));
			snprintf(pathbuf, 256, "%s/%d.status", blockpath.c_str(), i);
			muduo::string cp(pathbuf);
			BlockStatusPtr status = parse(cp, block, maxblocksize);
			table_.insert(std::pair<int, BlockStatusPtr>(i, status));
			size_.add(1);
		}
	}

	static BlockStatusPtr parse(const muduo::string &fullname, 
		BlockPtr block, long maxblocksize)
	{
		LOG_DEBUG << "read block config: " << fullname;
		sdfs::JsonValuePtr tree = sdfs::JsonHelper::ParseJson(fullname);
		if(tree == NULL)
		{
			sdfs::storage::BlockStatusPtr status(new BlockStatus(block->getIndex(), 
				block, maxblocksize));
			SaveStatus(fullname, status);
			return status;
		}
		int idx = (*tree)["id"].asInt();
		long capacity = (*tree)["blockcapacityMB"].asInt();//MB
		sdfs::storage::BlockStatusPtr status(new BlockStatus(idx, block, capacity*1024*1024));
		Json::Value pieces = (*tree)["freepieces"];
		long freesize = 0;
		for(uint32_t i = 0 ; i < pieces.size() ; ++i)
		{
			long offset= pieces[i]["offset"].asInt();
			long length = pieces[i]["port"].asInt();
			freesize += length;
			PiecePtr piece(new Piece(idx, offset, length));
			status->addFreePiece(piece);
		}
		status->setFreeSize(freesize);
		return status;
	}

	static void SaveStatus(const muduo::string& path, 
		const BlockStatusPtr status)
	{
		LOG_DEBUG << "Save Status";
		Json::Value root;
		root["id"] = status->getIndex();
		root["blockcapacityMB"] = (uint)status->getCapacity();
		root["freesize"] = (uint)status->getFreeSize();
		Json::Value freepiece;
		PieceTable* pieces = status->getPieces();
		PieceTable::iterator it;
		for (it = pieces->begin(); it != pieces->end(); it++)
		{
			Json::Value piece;
			piece["offset"] = (uint)(*it)->getOffset();
			piece["length"] = (uint)(*it)->getLength();
    			freepiece.append(piece);
		}
		root["freepiece"] = freepiece;
		JsonHelper::SerilizeJson(path, root);
	}



private:
	typedef std::map<int, BlockStatusPtr> BlockTable;
	BlockTable table_;
	AtomicInt32 index_;
	AtomicInt32 size_;
};

typedef boost::shared_ptr<BlockStatusTable> BlockStatusTablePtr;

}
}
#endif